package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.{SparkConf, SparkContext}

/**
  TODO 数据集的操作 -- Join
    1、Join算子可能会出现笛卡儿积成绩，也可能出现shuffle，可以用其他算子实现的话，尽量不要join
    2、可以将多个数据集一起进行连接 cogroup(多个数据集)，可以将同一个RDD的相同key的数据放在一起,就是为啥结果是CompactBuffer
    3、cogroup可以理解为：connect + group
 */

object Spark_RDD1_To_Rdd2_Pper_Transform_Join {
  def main(args: Array[String]): Unit = {


  //TODO 1、获得连接
  val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
  val sc: SparkContext = new SparkContext(conf)

    val rdd1 = sc.makeRDD(
      List(
       ("a",1),"b" -> 2,("c" ->3)
      ), 2
    )
    val rdd2 = sc.makeRDD(
      List(
        ("a",4),"b" -> 5,("c" ->6)
      ), 3
    )

    //这个可能出现笛卡尔积
    val rdd3 = sc.makeRDD(
      List(
        ("a",4),"a" -> 5,("a" ->6)
      ), 3
    )


      println("测试内连接" + rdd1.join(rdd2).collect().mkString(","))
      println("测试左外连接" + rdd1.leftOuterJoin(rdd2).collect().mkString(","))
      println("测试右外连接" + rdd1.rightOuterJoin(rdd2).collect().mkString(","))
      println("测试全连接" + rdd1.fullOuterJoin(rdd2).collect().mkString(","))
      println("测试迪卡尔" + rdd1.join(rdd3).collect().mkString(","))


    // cogroup可以将多个数据集一起进行连接
    // cogroup算子可以将同一个RDD的相同key的数据放一起
    // cogroup = connect + group
      println("测试多个数据集一起连接" + rdd1.cogroup(rdd3,rdd2,rdd1).collect().mkString(","))

  //TODO 2、关闭连接
  sc.stop()
  }
}